package cn.liyunqi.iot.chinatelecomunitimitate.compent.demo02;

import java.io.File;


import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;


/**
 *  MQTT 数据处理插件
 * @author chao
 * @createDate 2017.1.23
 */
public class McListener {
    private String file_conf = "mqtt.properties";

    private MqttClient client_sub;
    private MqttConnectOptions options_sub;
    private String topic_dev = "/examle/data/#";
    private String topic_online = "/examle/online/#";
    private String topic_will = "/examle/will/#";

    private static Logger Log = LoggerFactory.getLogger(McListener.class);

    private static final McListener listener = new McListener();

    public static McListener getInstance() {
        return listener;
    }

    private McListener() {

    }

    @Value("mqtt.tcp.host")
    private String mqttTcpHost;

    @Value("mqtt.clientid.subscribe")
    private String mqttClientidSubscribe;

    @Value("mqtt.username")
    private String mqttUsername;

    @Value("mqtt.password")
    private String mqttPassword;



    private void initMQTTListener() {
        try {
            //获取配置信息
            StringBuilder builder = new StringBuilder();
           // String filepath = builder.append(Constant.PATH_CONF).append(File.separator)
           //         .append(file_conf).toString();
            //PropertyUtil mqttProperty = new PropertyUtil(filepath);
            //mqttProperty.read();

            // HOST_MQ为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client_sub = new MqttClient(mqttTcpHost, mqttClientidSubscribe, new MemoryPersistence());
            // MQTT的连接设置
            options_sub = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
            options_sub.setCleanSession(false);
            // 设置连接的用户名
            options_sub.setUserName(mqttUsername);
            // 设置连接的密码
            options_sub.setPassword(mqttPassword.toCharArray());
            // 设置会话心跳时间 单位为秒 服务器会每隔90秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options_sub.setKeepAliveInterval(90);
            //订阅topic定义
            int[] Qos = new int[]{0, 0, 0};
            String[] topics = new String[]{topic_online, topic_will, topic_dev};

            // 设置回调
            client_sub.setCallback(new MqttCallbackExtended() {
                public void connectComplete(boolean reconnect, String serverURI) {
                    //连接成功，需要上传客户端所有的订阅关系
                    try {
                        client_sub.subscribe(topics, Qos);
                        Log.error("=======连接MQTT HOST 成功,重发Topics 1{}=2{}=3{}=======", topic_online, topic_will, topic_dev);
                    } catch (Exception e) {
                        Log.error("=======重连MQTT HOST 失败: {}, case: {}=========", serverURI, e.toString());
                    }
                }

                public void connectionLost(Throwable cause) {
                    // 连接丢失后，一般在这里面进行重连
                    Log.error("=======连接断开，可以做重连==============");
                    // reConnect();
                }

                public void deliveryComplete(IMqttDeliveryToken token) {
                    Log.error("=======交付完成: {}==============", token.isComplete());
                }

                public void messageArrived(String topic, MqttMessage message) {
                    //due arrived message...
                    Log.error("=======收到消息topic: {}===Qos: {}", topic, message.getQos());
                    Log.error("=======message: {}", message.toString());
                }
            });
            //连接mqtt服务器broker
            client_sub.connect(options_sub);
            //订阅消息
            client_sub.subscribe(topics, Qos);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            Log.error("init listener MQTT err info: {}", e.toString());
            System.exit(-1);
        }
    }

    /**
     * mqtt重连
     */
    public void reConnect() {
        try {
            if (null != client_sub && !(client_sub.isConnected())) {
                client_sub.reconnect();
                Log.error("=======尝试重新连接==============");
            }
        } catch (MqttException e) {
            Log.error("=======重新连接失败:{}==============", e.toString());
        }
    }

    public static void main(String[] args) {
        // 初始化数据侦听模块
        McListener.getInstance().initMQTTListener();
    }

    public void publish(String topic, MqttMessage message) throws MqttException, MqttPersistenceException {
        this.aClient.publish(topic, message, (Object)null, (IMqttActionListener)null).waitForCompletion(this.getTimeToWait());
    }

}